package com.sjgs.gis.fs;

import com.sjgs.common.utils.DirUtils;
import com.sjgs.gis.domain.Chunk;
import com.sjgs.gis.domain.Node;
import com.sjgs.gis.domain.enumeration.ChunkStatus;
import com.sjgs.gis.errors.FileGrpcSyncException;
import com.sjgs.gis.utils.GrpcChannelUtils;
import com.sjgs.gis.fstrans.*;
import com.sjgs.gis.service.ChunkService;
import com.sjgs.gis.errors.FileGrpcConnectException;
import com.sjgs.gis.errors.LocalFileIOException;
import io.grpc.Channel;
import io.grpc.ManagedChannel;
import org.apache.commons.io.FileUtils;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.cloud.netflix.eureka.EurekaDiscoveryClient;
import org.springframework.stereotype.Service;
import reactor.core.publisher.Mono;

import java.io.File;
import java.io.IOException;
import java.nio.file.*;
import java.util.*;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.stream.Collectors;

/**
 * 副本同步
 *
 * @author jxw
 */
@Service
public class ChunkSyncService {

    private static final Log logger = LogFactory.getLog(ChunkSyncService.class);
    private final static int MAX_MSG_SIZE = 5 * 1024 * 1024;

    @Autowired
    EurekaDiscoveryClient discoveryClient;

    @Autowired
    ChunkService chkservice;

    @Autowired
    Node currentNode;

    @Value("${filesystem.rootDir:/data}")
    private String rootDir;

    @Value("${grpc.server.port:8768}")
    private int port;

    /**
     * 同步远端文件片到本地:用于合并文件
     *
     * @param dir
     * @param fkey
     * @return
     */
    public Mono<Long> SyncDiffChunkToLocal(String fkey, String dir) {
        long finalStart = System.currentTimeMillis();
        String localAddr = currentNode.getInstanceId();
        String chunkDir = DirUtils.getFileDir(rootDir, dir, fkey);
        return chkservice.findAllByFkeyAndStatusAndChunkAddrIsNot(fkey, ChunkStatus.FINISHED.ordinal(), localAddr)
                .collectList()
                .flatMap(chunks -> {
                    logger.debug("findAllByFkeyAndStatusAndChunkAddrIsNot " + localAddr + " cost time :" + String.valueOf((System.currentTimeMillis() - finalStart)) + "ms");
                    int count = 0;
                    //按照存储节点分组
                    Map<String, List<Chunk>> groups = chunks.stream()
                            .collect(Collectors.groupingBy(Chunk::getChunkAddr));
                    Iterator<String> it = groups.keySet().iterator();
                    while (it.hasNext()) {
                        String target = it.next();
                        List<Chunk> target_chunks = groups.get(target)
                                .stream()
                                .sorted(Comparator.comparing(Chunk::getIndex))
                                .filter(chunk -> {
                                    Path chunkPath = Paths.get(chunkDir, "/", String.valueOf(chunk.getIndex()));
                                    if(Files.notExists(chunkPath)){
                                        return true;
                                    }else{
                                       return FileUtils.sizeOf(chunkPath.toFile()) != chunk.getSize();
                                    }
                                })
                                .collect(Collectors.toList());
                        if (target_chunks.size() > 0) {
                            int fetch_size = fetchDiffChunk(target, target_chunks, fkey, dir);
                            if (fetch_size != target_chunks.size()) {
                                throw new FileGrpcSyncException(target);
                            } else {
                                count += fetch_size;
                            }
                        }
                    }
                    return Mono.justOrEmpty(Long.valueOf(count));
                });
    }

    /**
     * 远程拉取差分片
     *
     * @param target
     * @param fkey
     * @param dir
     * @return
     */
    public Integer fetchDiffChunk(String target, List<Chunk> chunks, String fkey, String dir) {
        AtomicInteger count = new AtomicInteger();
        ManagedChannel channel = GrpcChannelUtils.getChannelByInstanceId(discoveryClient, currentNode.getServiceId(), target);
        if (channel == null) {
            throw new FileGrpcConnectException(target);
        }
        try {
            FSTransServiceGrpc.FSTransServiceBlockingStub transServiceStub = getClient(channel);
            String chunkdir = DirUtils.getFileDir(rootDir, dir, fkey);
            DirUtils.mkDir(chunkdir);
            for (Chunk chunk : chunks) {
                FetchChunkRequest request = FetchChunkRequest.newBuilder()
                        .setFkey(fkey)
                        .setIndex(chunk.getIndex())
                        .setFdir(dir)
                        .build();
                FetchChunkResponse fetchChunkResponse = transServiceStub.fetchChunk(request);
                if (!fetchChunkResponse.getChunk().isEmpty()) {
                    logger.info("async-" + fetchChunkResponse.getIndex() + ":" + fetchChunkResponse.getChunk().toByteArray().length);
                    Path chunkPath = DirUtils.getChunkPath(rootDir, dir, fkey, fetchChunkResponse.getIndex());
                    try {
                        Files.write(chunkPath, fetchChunkResponse.getChunk().toByteArray());
                        count.getAndIncrement();
                    } catch (IOException e) {
                        throw new LocalFileIOException("write Diff Chunk error : " + chunkPath.toString());
                    }
                }
            }
        }catch (Exception e){
            logger.error(e.getMessage());
        }finally {
            channel.shutdownNow();
        }
        return count.get();
    }

    /**
     * 创建阻塞客户端
     *
     * @param channel
     * @return
     */
    public FSTransServiceGrpc.FSTransServiceBlockingStub getClient(Channel channel) {
        return FSTransServiceGrpc.newBlockingStub(channel)
                .withMaxInboundMessageSize(MAX_MSG_SIZE);
    }
}
